第14章 RAG系统生成与评估
学习目标
- 掌握RAG系统中生成组件的设计与实现
- 学习如何优化检索结果与生成模型的协同
- 理解RAG系统评估的关键指标与方法
- 了解如何持续优化RAG系统性能
高质量生成系统实现
在RAG知识库系统中，生成组件负责将检索到的信息转化为连贯、准确的回答。生成质量直接决定了用户体验，因此需要精心设计。
1. 优化提示模板
设计专业的提示模板，引导大语言模型有效利用检索信息：
python
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.llms import DeepSeek
# 基础RAG提示模板
rag_basic_template = """
你是一个专业的AI助手。请基于以下检索到的信息，回答用户的问题。
如果检索到的信息不足以回答问题，请清楚地说明，不要编造信息。
检索到的信息:
{context}
用户问题: {question}
回答:
"""
# 高级RAG提示模板
rag_advanced_template = """
你是一个专业的AI知识库助手。你的任务是根据提供的上下文信息，回答用户的问题。
检索到的信息:
{context}
用户问题: {question}
在回答时，请遵循以下规则:
1. 仅使用提供的上下文信息回答问题，不要添加未在上下文中提及的信息
2. 如果上下文信息不足以完全回答问题，请明确指出信息的局限性
3. 保持回答的逻辑性和连贯性，不要简单拼接上下文中的句子
4. 使用客观、准确的语言，避免模糊或误导性表述
5. 回答应当条理清晰，易于理解
6. 如果合适，可以在回答最后提供参考出处
回答:
"""
# 支持自定义格式的RAG模板
rag_format_template = """
你是一个专业的AI知识库助手。你的任务是根据提供的上下文信息，回答用户的问题。
检索到的信息:
{context}
用户问题: {question}
输出格式: {output_format}
在回答时，请遵循以下规则:
1. 仅使用提供的上下文信息回答问题
2. 如果上下文信息不足以完全回答问题，请明确指出
3. 严格按照指定的输出格式组织回答
回答:
"""
# 创建提示模板
basic_prompt = PromptTemplate(
template=rag_basic_template,
input_variables=["context", "question"]
)
advanced_prompt = PromptTemplate(
template=rag_advanced_template,
input_variables=["context", "question"]
)
format_prompt = PromptTemplate(
template=rag_format_template,
input_variables=["context", "question", "output_format"]
)
# 创建LLM对象
llm = DeepSeek(api_key="your-api-key")
# 创建生成链
basic_chain = LLMChain(llm=llm, prompt=basic_prompt)
advanced_chain = LLMChain(llm=llm, prompt=advanced_prompt)
format_chain = LLMChain(llm=llm, prompt=format_prompt)2. 上下文组织与重排序
有效组织检索内容，提高生成质量：
python
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class ContextProcessor:
def __init__(self, llm, embeddings):
self.llm = llm
self.embeddings = embeddings
# 创建相关性评分链
relevance_template = """
评估以下文档与查询的相关性:
查询: {query}
文档: {document}
给出0-10的相关性评分，其中0代表完全不相关，10代表高度相关:
"""
self.relevance_scorer = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["query", "document"],
template=relevance_template
)
)
# 创建上下文组织链
organization_template = """
重新组织以下与查询相关的文本片段，形成一个连贯的上下文。
去除重复信息，按照逻辑顺序排列，确保上下文流畅。
查询: {query}
文本片段:
{documents}
组织后的上下文:
"""
self.context_organizer = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["query", "documents"],
template=organization_template
)
)
def rank_by_relevance(self, query, documents, method="vector"):
"""对文档按相关性排序"""
if method == "llm":
# 使用LLM评估相关性
scores = []
for doc in documents:
try:
score_text = self.relevance_scorer.run(
query=query,
document=doc.page_content[:1000] # 限制长度
)
score = float(score_text.strip())
scores.append((doc, score))
except:
scores.append((doc, 0))
# 按评分排序
sorted_docs = [doc for doc, _ in sorted(scores, key=lambda x: x[1], reverse=True)]
return sorted_docs
else:
# 使用向量相似度排序
query_embedding = self.embeddings.embed_query(query)
doc_embeddings = [self.embeddings.embed_document(doc.page_content) for doc in documents]
similarities = []
for i, doc_embedding in enumerate(doc_embeddings):
similarity = cosine_similarity(
np.array(query_embedding).reshape(1, -1),
np.array(doc_embedding).reshape(1, -1)
)[0][0]
similarities.append((documents[i], similarity))
sorted_docs = [doc for doc, _ in sorted(similarities, key=lambda x: x[1], reverse=True)]
return sorted_docs
def organize_context(self, query, documents, max_tokens=3000):
"""组织上下文"""
# 按相关性排序
ranked_docs = self.rank_by_relevance(query, documents)
# 组织上下文（文本长度限制）
context_text = ""
for doc in ranked_docs:
if len(context_text) + len(doc.page_content) > max_tokens:
break
context_text += doc.page_content + "\n\n"
# 如果文档很多或者很长，使用LLM组织上下文
if len(ranked_docs) > 3 or len(context_text) > 2000:
try:
documents_text = "\n\n".join([f"文档{i+1}: {doc.page_content[:500]}..."
for i, doc in enumerate(ranked_docs[:5])])
context_text = self.context_organizer.run(
query=query,
documents=documents_text
)
except Exception as e:
print(f"Error organizing context: {e}")
return context_text
# 实例化上下文处理器
context_processor = ContextProcessor(llm, embeddings)
# 处理检索结果
retrieved_docs = retrieval_system.retrieve("量子计算的基本原理和应用")
context = context_processor.organize_context("量子计算的基本原理和应用", retrieved_docs)3. 引用与可验证性
实现引用跟踪和可验证生成，提高回答的可信度：
python
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
class VerifiableRAGGenerator:
def __init__(self, llm):
self.llm = llm
# 创建可验证生成链
verifiable_template = """
你是一个专业的知识库助手。请基于提供的参考信息回答用户问题，并标注引用出处。
参考信息:
{context}
用户问题: {question}
请遵循以下规则:
1. 回答时，使用方括号标注引用，如[1]、[2]等
2. 回答末尾列出所有引用的来源
3. 不要编造未在参考信息中的内容
4. 如果参考信息不足以回答问题，请诚实说明
回答:
"""
self.verifiable_generator = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["context", "question"],
template=verifiable_template
)
)
# 创建源文档提取链
source_extraction_template = """
从以下文档中提取关于 "{question}" 的关键信息，并生成引用格式。
文档:
{document}
生成指定格式的引用内容:
"""
self.source_extractor = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["question", "document"],
template=source_extraction_template
)
)
def prepare_context_with_sources(self, query, documents):
"""准备带有源信息的上下文"""
source_contexts = []
for i, doc in enumerate(documents):
# 提取文档元数据
source_info = f"[{i+1}] "
if "source" in doc.metadata:
source_info += f"来源: {doc.metadata['source']} "
if "title" in doc.metadata:
source_info += f"标题: {doc.metadata['title']} "
if "date" in doc.metadata:
source_info += f"日期: {doc.metadata['date']} "
if "author" in doc.metadata:
source_info += f"作者: {doc.metadata['author']} "
# 提取内容
content = doc.page_content
# 组合为带源的上下文
source_contexts.append(f"{source_info}\n{content}")
return "\n\n".join(source_contexts)
def generate_with_citations(self, query, documents):
"""生成带引用的回答"""
# 准备带源的上下文
source_context = self.prepare_context_with_sources(query, documents)
# 生成带引用的回答
response = self.verifiable_generator.run(
context=source_context,
question=query
)
return response
# 实例化可验证RAG生成器
verifiable_generator = VerifiableRAGGenerator(llm)
# 生成带引用的回答
answer = verifiable_generator.generate_with_citations(
"量子计算的基本原理和应用",
retrieved_docs
)
print(answer)4. 多步生成与自我修正
实现多步思考和自我修正机制，提高生成质量：
python
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
class MultiStepRAGGenerator:
def __init__(self, llm):
self.llm = llm
# 第一步：信息分析链
analysis_template = """
你是一个专业的信息分析师。请分析以下与查询相关的上下文信息，提取关键点和解决方案。
查询: {query}
上下文信息:
{context}
请分析上下文信息，识别以下内容：
1. 关键事实和信息点
2. 与查询相关的核心概念
3. 可能的答案要点或解决方案
4. 信息中的任何缺口或不确定性
分析结果:
"""
self.analyzer = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["query", "context"],
template=analysis_template
)
)
# 第二步：草稿生成链
draft_template = """
你是一个专业的内容创作者。基于以下分析结果和原始上下文，创建一个回答草稿。
查询: {query}
分析结果:
{analysis}
原始上下文:
{context}
请创建一个初步的回答草稿，确保涵盖分析中的关键点:
"""
self.drafter = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["query", "analysis", "context"],
template=draft_template
)
)
# 第三步：自我修正链
revision_template = """
你是一个严格的内容审核员。请仔细检查以下回答草稿，并进行必要的修正。
查询: {query}
原始上下文:
{context}
回答草稿:
{draft}
请检查以下方面并进行修正:
1. 事实准确性 - 回答是否与上下文信息一致
2. 完整性 - 是否完整回答了查询
3. 清晰度 - 表述是否清晰易懂
4. 逻辑性 - 回答是否有逻辑结构
5. 客观性 - 是否避免了不必要的主观判断
修正后的最终回答:
"""
self.reviser = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["query", "context", "draft"],
template=revision_template
)
)
def generate(self, query, context):
"""执行多步生成"""
# 第一步：分析信息
analysis = self.analyzer.run(query=query, context=context)
# 第二步：生成草稿
draft = self.drafter.run(query=query, analysis=analysis, context=context)
# 第三步：修正改进
final_answer = self.reviser.run(query=query, context=context, draft=draft)
return {
"analysis": analysis,
"draft": draft,
"final_answer": final_answer
}
# 实例化多步生成器
multi_step_generator = MultiStepRAGGenerator(llm)
# 生成回答
result = multi_step_generator.generate(
"量子计算的基本原理和应用",
context
)
print("最终回答:")
print(result["final_answer"])RAG系统评估方法
评估RAG系统性能是优化系统的关键步骤。下面我们将实现多种评估方法：
1. 检索质量评估
评估检索组件的性能：
python
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
import numpy as np
class RetrievalEvaluator:
def __init__(self, llm):
self.llm = llm
# 创建相关性评估链
relevance_template = """
评估以下文档与查询的相关性:
查询: {query}
文档: {document}
请给出1-5的相关性评分，并解释原因:
1 = 完全不相关
2 = 略微相关
3 = 部分相关
4 = 高度相关
5 = 完全相关
评分和解释:
"""
self.relevance_evaluator = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["query", "document"],
template=relevance_template
)
)
def evaluate_retrieved_documents(self, query, documents, sample_size=None):
"""评估检索到的文档"""
if sample_size and len(documents) > sample_size:
# 随机抽样
np.random.seed(42) # 为了可重现性
sample_indices = np.random.choice(len(documents), sample_size, replace=False)
sample_docs = [documents[i] for i in sample_indices]
else:
sample_docs = documents
# 评估每个文档
evaluations = []
for doc in sample_docs:
try:
evaluation = self.relevance_evaluator.run(
query=query,
document=doc.page_content[:1000] # 限制文本长度
)
# 提取评分（简单解析，实际应用中可能需要更复杂的解析）
score = None
for line in evaluation.split('\n'):
if line.strip().isdigit() or (line.strip() and line.strip()[0].isdigit()):
try:
score = int(line.strip()[0])
break
except:
pass
if score is None:
score = 0
evaluations.append({
"document": doc.page_content[:100] + "...", # 文档摘要
"score": score,
"evaluation": evaluation
})
except Exception as e:
print(f"Error evaluating document: {e}")
evaluations.append({
"document": doc.page_content[:100] + "...",
"score": 0,
"evaluation": f"评估错误: {str(e)}"
})
# 计算统计指标
scores = [eval_item["score"] for eval_item in evaluations]
metrics = {
"mean_relevance": np.mean(scores) if scores else 0,
"median_relevance": np.median(scores) if scores else 0,
"min_relevance": min(scores) if scores else 0,
"max_relevance": max(scores) if scores else 0,
"highly_relevant_ratio": sum(1 for s in scores if s >= 4) / len(scores) if scores else 0
}
return {
"evaluations": evaluations,
"metrics": metrics
}
# 实例化检索评估器
retrieval_evaluator = RetrievalEvaluator(llm)
# 评估检索结果
retrieval_evaluation = retrieval_evaluator.evaluate_retrieved_documents(
"量子计算的基本原理和应用",
retrieved_docs,
sample_size=5
)
print("检索评估指标:")
for metric, value in retrieval_evaluation["metrics"].items():
print(f"{metric}: {value:.2f}")2. 生成质量评估
评估生成内容的质量：
python
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
class GenerationEvaluator:
def __init__(self, llm):
self.llm = llm
# 创建回答质量评估链
quality_template = """
评估以下AI回答的质量:
查询: {query}
原始上下文:
{context}
AI回答:
{answer}
请从以下维度评估回答质量，每个维度评分1-5分:
1. 相关性 - 回答与查询的相关程度
2. 准确性 - 回答与上下文事实的一致程度
3. 完整性 - 回答是否全面覆盖了查询的要点
4. 一致性 - 回答内部是否逻辑一致
5. 清晰度 - 回答是否清晰易懂
请给出每个维度的评分和简要解释:
"""
self.quality_evaluator = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["query", "context", "answer"],
template=quality_template
)
)
# 创建幻觉检测链
hallucination_template = """
检测以下AI回答中是否存在"幻觉"(即回答中包含上下文中不存在的信息):
查询: {query}
原始上下文:
{context}
AI回答:
{answer}
请仔细分析回答中的每个关键事实和声明，判断是否在上下文中有依据。
列出所有可能的幻觉内容，如果没有发现幻觉，请说明。
幻觉检测结果:
"""
self.hallucination_detector = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["query", "context", "answer"],
template=hallucination_template
)
)
def evaluate_answer(self, query, context, answer):
"""全面评估生成的回答"""
# 质量评估
try:
quality_evaluation = self.quality_evaluator.run(
query=query,
context=context[:3000], # 限制上下文长度
answer=answer
)
except Exception as e:
quality_evaluation = f"评估错误: {str(e)}"
# 幻觉检测
try:
hallucination_detection = self.hallucination_detector.run(
query=query,
context=context[:3000], # 限制上下文长度
answer=answer
)
except Exception as e:
hallucination_detection = f"检测错误: {str(e)}"
return {
"quality_evaluation": quality_evaluation,
"hallucination_detection": hallucination_detection
}
# 实例化生成评估器
generation_evaluator = GenerationEvaluator(llm)
# 评估生成的回答
answer_evaluation = generation_evaluator.evaluate_answer(
"量子计算的基本原理和应用",
context,
result["final_answer"]
)
print("回答质量评估:")
print(answer_evaluation["quality_evaluation"])
print("\n幻觉检测:")
print(answer_evaluation["hallucination_detection"])3. 基于真实答案的评估
当有标准答案时，可以进行更精确的评估：
python
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
class GroundTruthEvaluator:
def __init__(self, llm):
self.llm = llm
# 创建答案比较链
comparison_template = """
比较AI生成的回答与标准答案:
查询: {query}
AI生成的回答:
{generated_answer}
标准答案:
{ground_truth}
请从以下方面评估AI回答与标准答案的一致性:
1. 准确性 - AI回答中的事实是否与标准答案一致
2. 完整性 - AI回答是否涵盖了标准答案中的所有关键点
3. 信息增益 - AI回答是否提供了标准答案中没有的有价值信息
4. 表达效果 - AI回答的清晰度、结构性和可读性与标准答案相比如何
对于每个方面，给出1-5的评分和简要解释。然后给出总体评分(1-5)。
"""
self.answer_comparator = LLMChain(
llm=llm,
prompt=PromptTemplate(
input_variables=["query", "generated_answer", "ground_truth"],
template=comparison_template
)
)
def evaluate_with_ground_truth(self, query, generated_answer, ground_truth):
"""基于标准答案评估生成内容"""
comparison = self.answer_comparator.run(
query=query,
generated_answer=generated_answer,
ground_truth=ground_truth
)
return comparison
# 假设有标准答案
ground_truth_answer = """
量子计算是一种利用量子力学原理进行信息处理的计算方式。其基本原理基于量子比特（qubit）而非经典比特，量子比特可以同时处于多个状态的叠加，这使得量子计算机在处理某些问题时具有指数级的速度优势。
量子计算的核心原理包括:
1. 量子叠加态：量子比特可以同时存在于0和1的状态
2. 量子纠缠：多个量子比特间的特殊关联
3. 量子干涉：波函数的相消或相长干涉
主要应用领域包括:
- 密码学：破解现有加密系统和创建量子安全加密
- 药物研发：模拟分子结构和相互作用
- 优化问题：解决复杂的优化算法
- 机器学习：加速某些机器学习算法
- 金融建模：风险分析和投资组合优化
目前量子计算仍处于早期发展阶段，面临退相干、错误校正等技术挑战。
"""
# 使用标准答案评估
ground_truth_evaluator = GroundTruthEvaluator(llm)
gt_evaluation = ground_truth_evaluator.evaluate_with_ground_truth(
"量子计算的基本原理和应用",
result["final_answer"],
ground_truth_answer
)
print("与标准答案比较的评估:")
print(gt_evaluation)4. 系统整体评估与优化
进行端到端系统评估，识别优化方向：
python
class RAGSystemEvaluator:
def __init__(self, retrieval_system, generation_system, retrieval_evaluator, generation_evaluator):
self.retrieval_system = retrieval_system
self.generation_system = generation_system
self.retrieval_evaluator = retrieval_evaluator
self.generation_evaluator = generation_evaluator
def evaluate_system(self, queries, ground_truths=None):
"""评估整个RAG系统性能"""
system_evaluations = []
for i, query in enumerate(queries):
# 检索阶段
retrieved_docs = self.retrieval_system.retrieve(query)
retrieval_evaluation = self.retrieval_evaluator.evaluate_retrieved_documents(
query, retrieved_docs, sample_size=5
)
# 上下文处理
context = context_processor.organize_context(query, retrieved_docs)
# 生成阶段
generation_result = self.generation_system.generate(query, context)
answer = generation_result["final_answer"]
# 生成评估
generation_evaluation = self.generation_evaluator.evaluate_answer(
query, context, answer
)
# 如果有标准答案，进行基于标准答案的评估
ground_truth_evaluation = None
if ground_truths and i < len(ground_truths):
gt_evaluator = GroundTruthEvaluator(llm)
ground_truth_evaluation = gt_evaluator.evaluate_with_ground_truth(
query, answer, ground_truths[i]
)
# 合并评估结果
system_evaluations.append({
"query": query,
"retrieval_metrics": retrieval_evaluation["metrics"],
"generation_quality": generation_evaluation["quality_evaluation"],
"hallucination_detection": generation_evaluation["hallucination_detection"],
"ground_truth_evaluation": ground_truth_evaluation,
"answer": answer
})
# 计算系统整体指标
avg_retrieval_score = np.mean([eval_item["retrieval_metrics"]["mean_relevance"]
for eval_item in system_evaluations])
return {
"evaluations": system_evaluations,
"system_metrics": {
"avg_retrieval_relevance": avg_retrieval_score,
# 其他系统级指标可以根据需要添加
}
}
def identify_optimization_opportunities(self, system_evaluation):
"""识别系统优化机会"""
opportunities = {
"retrieval_issues": [],
"generation_issues": [],
"general_recommendations": []
}
# 分析检索问题
low_relevance_queries = []
for eval_item in system_evaluation["evaluations"]:
if eval_item["retrieval_metrics"]["mean_relevance"] < 3:
low_relevance_queries.append(eval_item["query"])
if low_relevance_queries:
opportunities["retrieval_issues"].append({
"issue": "低相关性检索结果",
"affected_queries": low_relevance_queries,
"recommendation": "优化检索策略，考虑使用查询扩展或混合检索方法"
})
# 添加更多优化建议...
return opportunities
# 评估系统性能
test_queries = [
"量子计算的基本原理和应用",
"深度学习在自然语言处理中的最新进展",
"区块链技术的优势和局限性"
]
# 系统评估
system_evaluator = RAGSystemEvaluator(
retrieval_system=retrieval_system,
generation_system=multi_step_generator,
retrieval_evaluator=retrieval_evaluator,
generation_evaluator=generation_evaluator
)
system_evaluation = system_evaluator.evaluate_system(test_queries)
optimization_opportunities = system_evaluator.identify_optimization_opportunities(system_evaluation)
print("系统评估结果:")
print(f"平均检索相关性: {system_evaluation['system_metrics']['avg_retrieval_relevance']:.2f}")
print("\n优化建议:")
for category, issues in optimization_opportunities.items():
print(f"\n{category}:")
for issue in issues:
print(f"- {issue['issue']}")
if 'affected_queries' in issue:
print(f" 影响的查询: {', '.join(issue['affected_queries'])}")
print(f" 建议: {issue['recommendation']}")思考题
在RAG系统中，生成组件与检索组件的质量哪个更重要？在哪些场景下这种重要性关系会发生变化？
如何处理RAG系统中的"幻觉"问题？有哪些技术和策略可以减少生成内容中的错误信息？
对于需要深度推理的复杂问题，标准RAG架构可能存在哪些局限性？如何改进？
在实际应用场景中，如何平衡RAG系统的生成质量和响应速度？有哪些优化策略？
设计一个自动化的RAG系统持续优化机制，它应该包含哪些关键组件和流程？